package ontime

import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time

/**
  * @author dinghao 
  * @create 2021-07-28 10:10 
  * @message
  */
object Timeflink {
  /*
dd 7325234234 2
dd 7325111234 1
dd 7325646734 6
dd 7325546734 5
dd 7325334234 3
dd 7325411234 4
  * */
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    env.setParallelism(1)
    val socketData = env.socketTextStream(args(0), 9999)
    val waterData = socketData.map(a => {
      val data = a.split(" ")
      (data(0), data(1).toLong, data(2).toInt)
    }).assignTimestampsAndWatermarks(
      new BoundedOutOfOrdernessTimestampExtractor[(String, Long, Int)](Time.seconds(0)) {
        override def extractTimestamp(element: (String, Long, Int)): Long = {
          element._2 * 1000
        }
      })

    val laterData = new OutputTag[(String, Long, Int)]("later")
    val result = waterData.keyBy(_._1)
      .window(TumblingEventTimeWindows.of(Time.seconds(5)))
      .sideOutputLateData(laterData)
      .reduce((a, b) => (a._1, math.max(a._2, b._2), a._3))
    waterData.print("data")
    result.print("result")
    result.getSideOutput(laterData).print("later")
    env.execute("Timeflink")
  }
}
